package top.lyc829.iot_android.mqtt;

import com.blankj.utilcode.util.SPUtils;

import org.eclipse.paho.android.service.MqttAndroidClient;
import org.eclipse.paho.client.mqttv3.DisconnectedBufferOptions;
import org.eclipse.paho.client.mqttv3.IMqttActionListener;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.IMqttToken;
import org.eclipse.paho.client.mqttv3.MqttCallbackExtended;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;

import java.util.LinkedHashMap;
import java.util.Map;
import java.util.UUID;

import top.lyc829.iot_android.MyApp;
import top.lyc829.iot_android.util.LogUtil;
import top.lyc829.iot_android.util.Utils;

/**
 * @Author lyc
 * @create 2021-06-19 15:08
 */
public class MqttManager {

    private final String serverUrl = "tcp://lyc829.top:1883";
    private String clientId = UUID.randomUUID().toString();

    private MqttAndroidClient mqttClient;

    private LinkedHashMap<String, Callback> callbackMap = new LinkedHashMap<>();

    private MqttManager() {
    }


    /**
     * 初始化
     */
    public void init() {
        String id = SPUtils.getInstance().getString("client_id", "-1");
        clientId = "-1".equals(id) ? clientId : id;
        mqttClient = new MqttAndroidClient(MyApp.getContext(), serverUrl, clientId);

        mqttClient.setCallback(new MqttCallbackExtended() {
            @Override
            public void connectComplete(boolean reconnect, String serverURI) {
                if (reconnect) {
                    LogUtil.SoutMqttd("Reconnected to : " + serverURI);
                    // Because Clean Session is true, we need to re-subscribe
                    subscribeToTopic();
                } else {
                    LogUtil.SoutMqttd("Connected to: " + serverURI);
                }

            }

            @Override
            public void connectionLost(Throwable cause) {
                LogUtil.SoutMqtte("The Connection was lost.");
            }

            @Override
            public void messageArrived(String topic, MqttMessage message) {

                Callback callback = callbackMap.get(topic);
                if (callback != null){
                    callback.callback(message);
                }

                LogUtil.SoutMqttd("MessageArrived Topic: " + topic + " Payload: " + new String(message.getPayload()));
            }

            @Override
            public void deliveryComplete(IMqttDeliveryToken token) {
                LogUtil.SoutMqttd("Delivery complete");
            }
        });

    }

    public void connect() {
        MqttConnectOptions options = new MqttConnectOptions();
        options.setAutomaticReconnect(true);
        options.setCleanSession(true);
        options.setConnectionTimeout(30);
        options.setKeepAliveInterval(60);

        try {
            mqttClient.connect(options, this, new IMqttActionListener() {
                @Override
                public void onSuccess(IMqttToken asyncActionToken) {
                    DisconnectedBufferOptions disconnectedBufferOptions = new DisconnectedBufferOptions();
                    disconnectedBufferOptions.setBufferEnabled(true);
                    disconnectedBufferOptions.setBufferSize(100);
                    disconnectedBufferOptions.setPersistBuffer(false);
                    disconnectedBufferOptions.setDeleteOldestMessages(false);
                    mqttClient.setBufferOpts(disconnectedBufferOptions);

                    subscribeToTopic();
                }

                @Override
                public void onFailure(IMqttToken asyncActionToken, Throwable exception) {
                    LogUtil.SoutMqttd("Failed to connect to: " + serverUrl);
                }
            });
        } catch (MqttException e) {
            e.printStackTrace();
        }
    }

    private void subscribeToTopic() {
        try {
            if (mqttClient != null) {
                String[] topics = callbackMap.keySet().toArray(new String[0]);
                int[] qos = new int[topics.length];
                mqttClient.subscribe(topics, qos, this, new IMqttActionListener() {
                    @Override
                    public void onSuccess(IMqttToken asyncActionToken) {
                        LogUtil.SoutMqttd(" subscribed!");
                    }

                    @Override
                    public void onFailure(IMqttToken asyncActionToken, Throwable exception) {
                        LogUtil.SoutMqttd("failed to subscribe");
                    }
                });

            } else {
                LogUtil.SoutMqtte( "mqttClient is null");
            }

        } catch (MqttException e) {
            e.printStackTrace();
        }
    }

    /**
     * 增加一个订阅
     * @param topic 主题
     * @param callback 回调接口
     */
    public void addSubscribe(String topic, Callback callback) {
        if (callbackMap.get(topic) == null) {
            callbackMap.put(topic,callback);

            // 等待mqtt连接上再订阅
            new Thread(() -> {
                while (true) {
                    if (mqttClient != null && mqttClient.isConnected()) {
                        subscribeToTopic();
                        break;
                    }
                }
            });
        }

    }

    private void clearSubscribes() {
        try {
            String[] topics = (String[]) callbackMap.keySet().toArray();
            mqttClient.unsubscribe(topics, this, new IMqttActionListener() {
                @Override
                public void onSuccess(IMqttToken asyncActionToken) {
                    LogUtil.SoutMqttd( "unsubscribed success");
                }

                @Override
                public void onFailure(IMqttToken asyncActionToken, Throwable exception) {
                    LogUtil.SoutMqtte( "unsubscribed failed");
                }
            });
        } catch (MqttException e) {
            e.printStackTrace();
        }
    }


    /**
     * 退出时调用
     */
    public void destroy() {
        if (mqttClient != null && mqttClient.isConnected()) {
            //clearSubscribes();
            try {
                mqttClient.disconnect();
            } catch (MqttException e) {
                e.printStackTrace();
            }
            mqttClient = null;
        }
    }


    /**
     * 发送一个消息
     * @param topic 主题
     * @param content 内容
     */
    public void publishMessage(String topic,String content) {

//        // 等待mqtt连接上再发布
//        new Thread(() -> {
//            while (true) {
//                if (mqttClient != null && mqttClient.isConnected()) {
//                    subscribeToTopic();
//                    break;
//                }
//            }
//        });

        if(mqttClient == null || !mqttClient.isConnected()) {
            return;
        }

        try {
            MqttMessage message = new MqttMessage();
            message.setQos(0);
            message.setPayload(content.getBytes());

            mqttClient.publish(topic, message, this, new IMqttActionListener() {
                @Override
                public void onSuccess(IMqttToken asyncActionToken) {
                    LogUtil.SoutMqttd(topic + ", message published");
                }

                @Override
                public void onFailure(IMqttToken asyncActionToken, Throwable exception) {
                    LogUtil.SoutMqtte("Failed to publish  message, " + exception.getMessage());
                }
            });
            if (!mqttClient.isConnected()) {
                LogUtil.SoutMqtte(mqttClient.getBufferedMessageCount() + " messages in buffer.");
            }
        } catch (MqttException e) {
            e.printStackTrace();
        }
    }


    public static MqttManager getInstance() {
        return INSTANCE.mqttManager;
    }

    private static class INSTANCE {
        private static final MqttManager mqttManager = new MqttManager();
    }

    public interface Callback{
        void callback(MqttMessage mqttMessage);
    }

    public interface Connect {
        void connectComplete();
    }

}
